package mapReduce.demo05_shuffle.demo02_phoneTrafficSort;

import mapReduce.util.JobSubmit;
import mapReduce.writable.PhoneTrafficWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * 原始数据：把手机流量求和的结果文件当成此案例的原始数据
 * 需求：按照手机流量的总流量倒序排序
 *
 * 思路：因为需要用到自定义序列化类型中的字段进行排序，所以这个自定义序列化类型必须为key，否则无法排序
 *
 */
public class PhoneTrafficSortApp {

    public static class PTSMapper extends Mapper<LongWritable, Text, PhoneTrafficWritable,Text>{

        PhoneTrafficWritable outKey = new PhoneTrafficWritable();
        Text outValue = new Text();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, PhoneTrafficWritable, Text>.Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");

            String phoneNum = fields[0];
            int upTrafficSum = Integer.parseInt(fields[1]);
            int downTrafficSum = Integer.parseInt(fields[2]);
            int sumTrafficSum = Integer.parseInt(fields[3]);

            outKey.setUpTraffic(upTrafficSum);
            outKey.setDownTraffic(downTrafficSum);
            outKey.setSumTraffic(sumTrafficSum);

            outValue.set(phoneNum);

            context.write(outKey,outValue);
        }
    }

    public static class PTSReducer extends Reducer<PhoneTrafficWritable,Text,Text, NullWritable>{

        Text outKey = new Text();

        @Override
        protected void reduce(PhoneTrafficWritable key, Iterable<Text> values, Reducer<PhoneTrafficWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {

            /**
             * 数据格式
             *      138xxxxxxxx 20  30  50
             *      139xxxxxxxx 20  30  50
             * 接收的数据格式
             *      20  30  50  {138xxxxxxxx,139xxxxxxxx}
             * 只需要遍历value，每遍历一个手机号分别与key组合
             */
            for (Text value : values) {
                outKey.set(value.toString()+"\t"+key.toString());
                context.write(outKey,NullWritable.get());
            }
        }
    }

    public static void main(String[] args) throws Exception {
        JobSubmit.submitBaseJob(PhoneTrafficSortApp.class,args,PhoneTrafficSortComparator.class);
    }

}
